package cn.zw.frr.ringbuffer;

import cn.zw.frr.factory.ValueWrapper;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadFactory;


public abstract class BaseCommond<D, E extends ValueWrapper<D>, H extends WorkHandler<E>> {

    private static List<BaseCommond> queueHelperList = new ArrayList<BaseCommond>();

    public D dto;

    public int count = 5;

    public Integer iBaseServiceNum;

    public volatile boolean CAN_USE = false;


    public void setiBaseServiceNum(Integer iBaseServiceNum) {
        this.iBaseServiceNum = iBaseServiceNum;
    }

    public BaseCommond setDTO(D dto){
        this.dto = dto;
        return this;
    };

    public BaseCommond go(){
        this.init();
        return this;
    }

    public BaseCommond setCount(int count) {
        this.count = count;
        return this;
    }


    private Disruptor<E> disruptor;

    private RingBuffer<E> ringBuffer;

    private List<D> initQueue = new ArrayList<D>();


    protected abstract int getQueueSize();


    protected abstract com.lmax.disruptor.EventFactory eventFactory();


    protected abstract WorkHandler[] getHandler(int count);


    public abstract void setMethodName(String methodName);

    public abstract void setParam(Class param);


    protected abstract WaitStrategy getStrategy();



    public void init() {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("DisruptorThreadPool").build();
        disruptor = new Disruptor<>(
                eventFactory(),
                getQueueSize(),
                namedThreadFactory,
                ProducerType.SINGLE,
                getStrategy());
        disruptor.setDefaultExceptionHandler(new HandlerException());
        disruptor.handleEventsWithWorkerPool(getHandler(count));
        ringBuffer = disruptor.start();

        //初始化数据发布
        for (D data : initQueue) {
            ringBuffer.publishEvent(new EventTranslatorOneArg<E, D>() {
                @Override
                public void translateTo(E event, long sequence, D data) {
                    event.setValue(data);
                }
            }, data);
        }

        //加入资源清理钩子
        synchronized (queueHelperList) {
            if (queueHelperList.isEmpty()) {
                Runtime.getRuntime().addShutdownHook(new Thread() {
                    @Override
                    public void run() {
                        for (BaseCommond baseHelper : queueHelperList) {
                            baseHelper.shutdown();
                        }
                    }
                });
            }
            queueHelperList.add(this);
        }
    }


    public synchronized void exec(D data) {
        System.out.println(":::::"+data);
        if (ringBuffer == null) {
            initQueue.add(data);
            return;
        }
        ringBuffer.publishEvent(new EventTranslatorOneArg<E, D>() {
            @Override
            public void translateTo(E event, long sequence, D data) {
                event.setValue(data);
            }
        }, data);
    }



    public void exec(){
        exec(this.dto);
    }


    public void shutdown() {
        disruptor.shutdown();
    }
}